package com.lvlian;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.utils.JdbcUtils;
import com.utils.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.HashMap;
import java.util.Map;
public class LvlianSinkPaymentInfo extends RichSinkFunction<String> {
    private static final Logger LOG = LoggerFactory.getLogger(LvlianSinkPaymentInfo.class);
    private Connection connection  ;
    private PreparedStatement preparedStatementInsert;
    private PreparedStatement preparedStatementDelete;
    private String lastValue = null;
    private Map<Integer, KV> map = new HashMap<>();

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        connection =    JdbcUtils.getInstance().getConnection();
        // 创建 PreparedStatement 对象
        String sql = "replace INTO lv_payment_info ( " +
                "id,payment_num,payment_type,pay_time,payer_name," +
                "payer_bank,payer_accno,payee_name,payee_bank,payee_accno," +
                "currency,amount,amount_words,channel,summary," +
                "bank_name,remain_amount,use_sts,is_advance_payment,is_with_invoice," +
                "create_user,create_date,org_code,lfrk_remain_amount,zsrk_remain_amount"
                + ") VALUES (" +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?" +
                ")";
        LOG.info(sql );
        preparedStatementInsert = connection.prepareStatement(sql);


        map.put(1, new KV("ROW_ID", "id"));
        map.put(2, new KV("PAYMENT_NUM", "payment_num"));
        map.put(3, new KV("PAYMENT_TYPE", "payment_type"));
        map.put(4, new KV("PAY_TIME", "pay_time"));
        map.put(5, new KV("PAYER_NAME", "payer_name"));
        map.put(6, new KV("PAYER_BANK", "payer_bank"));
        map.put(7, new KV("PAYER_ACCNO", "payer_accno"));
        map.put(8, new KV("PAYEE_NAME", "payee_name"));
        map.put(9, new KV("PAYEE_BANK", "payee_bank"));
        map.put(10, new KV("PAYEE_ACCNO", "payee_accno"));
        map.put(11, new KV("CURRENCY", "currency"));
        map.put(12, new KV("AMOUNT", "amount"));
        map.put(13, new KV("AMOUNT_WORDS", "amount_words"));
        map.put(14, new KV("CHANNEL", "channel"));
        map.put(15, new KV("ABSTRACT", "summary"));
        map.put(16, new KV("BANK_NAME", "bank_name"));
        map.put(17, new KV("REMAIN_AMOUNT", "remain_amount"));
        map.put(18, new KV("USE_STS", "use_sts"));
        map.put(19, new KV("IS_ADVANCE_PAYMENT", "is_advance_payment"));
        map.put(20, new KV("IS_WITH_INVOICE", "is_with_invoice"));
        map.put(21, new KV("RECORDER", "create_user"));
        map.put(22, new KV("RECORD_DATE", "create_date"));
        map.put(23, new KV("ORG_CODE", "org_code"));
        map.put(24, new KV("LFRK_REMAIN_AMOUNT", "lfrk_remain_amount"));
        map.put(25, new KV("ZSRK_REMAIN_AMOUNT", "zsrk_remain_amount"));


    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 将 JSON 字符串解析为 JSONObject
        JSONObject jsonObject = JSON.parseObject(value);
        // 获取操作类型
        String operation = jsonObject.getString("op");

        // 判断操作类型
        if ("d".equals(operation)) {
            // 删除操作
            JSONObject beforeObject = jsonObject.getJSONObject("before");
            String id = beforeObject.getString("ROW_ID");
            preparedStatementDelete.setString(1, id);
            preparedStatementDelete.executeUpdate();
        } else {
            // 插入或更新操作
            JSONObject afterObject = jsonObject.getJSONObject("after");
        // 如果上一次处理的值为空，或者当前值与上一次处理的值不相同，则执行 sink 操作
        if (lastValue == null || !lastValue.equals(value)) {
            // 遍历原始表字段
            for (Map.Entry<Integer, KV> entry : map.entrySet()) {
                String v = afterObject.getString(entry.getValue().getK());


                if("PAY_TIME".equals( entry.getValue().getK() )  ||  "RECORD_DATE".equals( entry.getValue().getK() )  ){
                    LOG.info(entry.getKey() + ": " + entry.getValue().getK() + " -> " + entry.getValue().getV()   + " -> " +   v  );
                    v = StringUtils.restoreDate(v);
                    LOG.info(entry.getKey() + ":处理后-============== " + entry.getValue().getK() + " -> " + entry.getValue().getV()   + " -> " +   v  );
                }

                preparedStatementInsert.setString(entry.getKey(), v);
            }

            preparedStatementInsert.executeUpdate();

            // 更新上一次处理的值为当前值
            lastValue = value;
        }
    }   }


    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接和 PreparedStatement
        if (preparedStatementInsert != null) {
            preparedStatementInsert.close();
        }
        if (preparedStatementDelete != null) {
            preparedStatementDelete.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
